package cn.doitedu.kafka.demos;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class Producer事务控制DEMO {

    public static void main(String[] args) {


        Properties props = new Properties();
        /**
         * 必配参数
         */
        //设置kafka集群的地址
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "doit01:9092,doit02:9092,doit03:9092");
        //序列化器
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // ack应答级别
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        //失败重试次数（有可能会造成数据的乱序）
        props.put("retries", 3);
        // in flight请求数最大上限
        props.setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION,"5");

        // 开启幂等性
        props.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,"true");

        // 指定事务id
        props.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"trans007");

        //数据发送的批次大小
        props.put("batch.size", 10);
        //生产者一次请求向服务器发送的数据量的最大字节数
        props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG,204800);
        //消息在缓冲区保留的时间，超过设置的值就会被提交到服务端
        props.put("linger.ms", 5000);
        //buffer.memory要大于batch.size，否则会报申请内存不足的错误
        props.put("buffer.memory", 102400000);  // 默认32M

        // 构造一个生产者实例对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        // 初始化一个事务
        producer.initTransactions();

        // 开始事务
        producer.beginTransaction();

        try {
            for (int i = 0; i < 100; i++) {
                if (i % 3 == 0) {
                    producer.send(new ProducerRecord<>("demo3", i + ""));
                    producer.flush();
                }
                if(i==77) throw new RuntimeException("哈哈哈，逗你玩");
            }

            producer.commitTransaction();

        }catch (Exception e){
            System.out.println("事务被放弃..........");
            producer.abortTransaction();
        }

        producer.close();
    }

}
